package com.navi.loaddata.job;

import com.navi.loaddata.core.etlException.ETLException2;
import com.navi.loaddata.core.gpEntity.base.ReloadPanel;
import com.navi.loaddata.core.mysqlEntity.EtlConfD;
import com.navi.loaddata.core.mysqlEntity.EtlConfDDefect;
import com.navi.loaddata.core.mysqlEntity.EtlConfDMapper;
import com.navi.loaddata.core.mysqlEntity.EtlConfDModel;
import com.navi.loaddata.core.util.JacksonUtil;
import com.navi.loaddata.core.util.LoggerUtil;
import com.navi.loaddata.loadService.IEtlConfServices2;
import com.navi.loaddata.loadService.impl.DefectData.LoadDefectDataBean;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;
import com.navi.loaddata.loadService.impl.DefectData.LoadDefectData_CService;
import com.navi.loaddata.email.MailService;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import com.navi.loaddata.loadService.SpringContext;

import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class LoadDefectData_C implements Job {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private IEtlConfServices2 etlConfServices;

    @Autowired
    private LoadDefectData_CService LoadDefectData_CService;

    @Autowired
    private LoadDefectDataBean loadDefectDataBean;


    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        //get JOB
        EtlConfD etlConfD = etlConfServices.getNewConf(context);
        if (null == etlConfD) return;

        List<EtlConfDDefect> etlHisJobList = loadDefectDataBean.getHisJobFlag(etlConfD);

        try {
            //判断如果从etl_conf_d表中查询对应his job的ext_2值为Y，则不做后续处理
            if (etlHisJobList.size() > 0) {
                for (EtlConfDDefect etlHisJob : etlHisJobList) {
                    if ("Y".equals(etlHisJob.getExt_2())) {
                        //跑current job，若runEndTimeStamp与nowTime的时间差大于BufferTime，将Ext_2置成Y，供His job做参考
                        if (etlConfD.getRunEndTimestamp() != null) {
                            Timestamp nowTime = Timestamp.valueOf(LocalDateTime.now());
                            long duration = (nowTime.getTime() - etlConfD.getRunEndTimestamp().getTime()) / 1000 / 60;
                            if ((int) duration >= (etlConfD.getBufferMinute() + etlConfD.getQueryMinute())) {
                                etlConfD.setExt2("Y");
                            } else {
                                etlConfD.setExt2("N");
                            }
                        } else {
                            etlConfD.setExt2("Y");
                        }
                        etlConfD.setExt3("His job is running");
                        loadDefectDataBean.updateEtlConfD(etlConfD);

                        String[] resMsg = {"0", "0", null};
                        EtlConfDModel etlConfDModel = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD);
                        etlConfServices.addHisOfModel(etlConfDModel, resMsg);

                        return;
                    } else {
                        etlConfD.setExt3("");

                        //跑current job，若runEndTimeStamp与nowTime的时间差大于BufferTime，将Ext_2置成Y，供His job做参考
                        if (etlConfD.getRunEndTimestamp() != null) {
                            Timestamp nowTime = Timestamp.valueOf(LocalDateTime.now());
                            long duration = (nowTime.getTime() - etlConfD.getRunEndTimestamp().getTime()) / 1000 / 60;
                            if ((int) duration >= (etlConfD.getBufferMinute() + etlConfD.getQueryMinute())) {
                                etlConfD.setExt2("Y");
                            } else {
                                etlConfD.setExt2("N");
                            }
                        } else {
                            etlConfD.setExt2("Y");
                        }

                        loadDefectDataBean.updateEtlConfD(etlConfD);
                    }
                }
            }

            etlConfD = etlConfServices.beginJob(etlConfD);
            if (StringUtils.isEmpty(etlConfD.getQueryMinute())) return;
            startETLByOpes(etlConfD);
        } catch (Exception e) {
            LoggerUtil.stackTraceToString(e);
        }
    }

    private void startETLByOpes(EtlConfD etlConfD) {
        //获取站点
        EtlConfDModel etlConfDModel = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD);

        List<String> errorOpeList = new ArrayList<>();

//        List<String> opeList = LoadDefectData_CService.getOpeList(etlConfDModel);
//        if(opeList.isEmpty()) return;//查询Opelist异常
        List<String> opeList = etlConfServices.getOpesList(etlConfD);

        if (null == opeList.get(0)) {//没有数据
            String[] resMsg = {"0", "0", null};//okCnt,errCnt,errMsg
            EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
            etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
            etlConfServices.endJob(etlConfD);
            return;
        }

        long loopCnt = 0;
        //循环OPE
        for (String opeid : opeList) {
            etlConfDModel.setOpe(opeid);//跑单站点
            String[] resMsg = new String[3];//okCnt,errCnt,errMsg
            try {
                resMsg = LoadDefectData_CService.loadDefectDataMain(etlConfDModel, etlConfD);
                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
                if (null != resMsg[2] && resMsg[2].indexOf("deadlock") > -1) return;
                if (null == etlConfDByOpe) return;
                etlConfDByOpe.setOpe(opeid);
                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
            } catch (ETLException2 e) {
                String err = e.getErrMsg();
                //print Log
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(), opeid,
                        e.getErrorCode().getErrorMsg(), LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (99999 == e.getErrorCode().getErrorCode()) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            } catch (Exception e) {
                String err = LoggerUtil.stackTraceToString(e);
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(), opeid,
                        LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (null != e && "Y".equals(etlConfServices.specialErr(etlConfDByOpe.getJobName(), err))) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            }
        }
        etlConfServices.endJob(etlConfD);
        etlConfD.setExt2("N");
        loadDefectDataBean.updateEtlConfD(etlConfD);

        //        //跑current job，若runEndTimeStamp与nowTime的时间差大于BufferTime，将Ext_2置成Y，供His job做参考
        //        if (etlConfD.getRunEndTimestamp() != null) {
        //            Timestamp nowTime = Timestamp.valueOf(LocalDateTime.now());
        //            long duration = (nowTime.getTime() - etlConfD.getRunEndTimestamp().getTime())/1000/60;
        //            if ((int)duration >= (etlConfD.getBufferMinute() + etlConfD.getQueryMinute())) {
        //                etlConfD.setExt2("Y");
        //            } else {
        //                etlConfD.setExt2("N");
        //            }
        //        } else {
        //            etlConfD.setExt2("Y");
        //        }
        //        loadDefectDataBean.updateEtlConfD(etlConfD);
        logger.info("loadDefectData_C run completed!");
    }

    public void loadDefectDataC(EtlConfD etlConfD, EtlConfDModel etlConfDModel, List<String> opeList, List<String> errorOpeList, String start_evt_timestamp, String end_evt_timestamp) {

    }

    private void errorCatch(EtlConfD etlConfD, long loopCnt, String opeid, List<String> errorOpeList, String errorLog) {
        //获取加入到opeList中的重复站点数量
        loopCnt = errorOpeList.stream().filter(s -> s.equals(opeid)).count();

        //报错站点数据重跑
        errorOpeList.add(opeid);

        if (loopCnt <= 4) {
            //获取站点
            EtlConfDModel etlConfDModel = EtlConfDMapper.INSTANCE.etlConfDToModel(etlConfD);

            etlConfDModel.setOpe(opeid);//跑单站点
            String[] resMsg = new String[3];//okCnt,errCnt,errMsg
            try {
                resMsg = LoadDefectData_CService.loadDefectDataMain(etlConfDModel, etlConfD);
                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
                if (null != resMsg[2] && resMsg[2].indexOf("deadlock") > -1) return;
                if (null == etlConfDByOpe) return;
                etlConfDByOpe.setOpe(opeid);
                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
            } catch (ETLException2 e) {
                String err = e.getErrMsg();
                //print Log
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(),opeid,
                        e.getErrorCode().getErrorMsg(), LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
                if (null == etlConfDByOpe) return;
                etlConfDByOpe.setOpe(opeid);
                resMsg[0] = "0";
                resMsg[1] = "-1";
                resMsg[2] = err;
                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);

                if (99999 == e.getErrorCode().getErrorCode()) {//无法连接,SLEEP
                    etlConfServices.resetJob(etlConfD);
                    return;
                }
            } catch (Exception e) {
                String err = LoggerUtil.stackTraceToString(e);
                logger.error("Job {} [{}]-[{}] OPE={} FAIL TO SYNCHRONIZE.\n{}",
                        etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName(),
                        etlConfD.getRunStartTimestamp(), etlConfD.getRunEndTimestamp(),opeid,
                        LoggerUtil.stackTraceToString(e)
                );

                errorCatch(etlConfD, loopCnt, opeid, errorOpeList, LoggerUtil.stackTraceToString(e));

//                EtlConfDModel etlConfDByOpe = etlConfServices.endJobOfModel(etlConfD);
//                if (null == etlConfDByOpe) return;
//                etlConfDByOpe.setOpe(opeid);
//                resMsg[0] = "0";
//                resMsg[1] = "-1";
//                resMsg[2] = err;
//                etlConfServices.addHisOfModel(etlConfDByOpe, resMsg);
//
//                if (null != e && "Y".equals(etlConfServices.specialErr(etlConfDByOpe.getJobName(), err))) {//无法连接,SLEEP
//                    etlConfServices.resetJob(etlConfD);
//                    return;
//                }
                return;
            }
        }else if (loopCnt > 4) {
            try {
                MailService mail = new MailService();
                mail.sendHtmlMail("chengyang.fu@s-navimentum.com", "DefectSummary同一站点报错超过4次", etlConfD.getProjectName() + "_" + etlConfD.getJobGroupName() + "_" + etlConfD.getJobName()
                        + "_" + etlConfD.getRunStartTimestamp() + "_" + etlConfD.getRunEndTimestamp() + "_" + opeid + "_" +
                        errorLog);
            } catch (Exception ex) {
                LoggerUtil.stackTraceToString(ex);
            }
        }
    }
}
